package org.example.mq.brokerserver;

import com.fasterxml.jackson.core.JsonProcessingException;
import org.example.mq.brokerserver.core.*;
import org.example.mq.brokerserver.core.ConsumerManager;
import org.example.mq.brokerserver.datacenter.DiskDataCenter;
import org.example.mq.brokerserver.datacenter.MemoryDataCenter;
import org.example.mq.common.Consumer;
import org.example.mq.common.ConsumerEnv;
import org.example.mq.common.MqException;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

//这个类表示虚拟主机
//每个主机都有自己的交换机，队列，绑定，消息数据
//一个虚拟主机就相当于一个主机，一个消息队列有多个虚拟主机（这里只使用一个作为实例）
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    //转发规则
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);
    //操作交换机的锁对象
    private final Object exchangeLocker = new Object();
    //操作队列的锁对象
    private final Object queueLocker = new Object();

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public void setVirtualHostName(String virtualHostName) {
        this.virtualHostName = virtualHostName;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public void setDiskDataCenter(DiskDataCenter diskDataCenter) {
        this.diskDataCenter = diskDataCenter;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public void setMemoryDataCenter(MemoryDataCenter memoryDataCenter) {
        this.memoryDataCenter = memoryDataCenter;
    }
    public VirtualHost(String name){
        this.virtualHostName=name;
        //初始化（这里只有Disk需要初始化）
        try {
            diskDataCenter.init();
        } catch (JsonProcessingException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 硬盘初始化失败！");
        }
        //将硬盘的数据恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (IOException | MqException | ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败！");
        }
    }
    //创建交换机
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable,
                                   boolean autoDelete, Map<String,Object> argument){
        //为了区分不同虚拟主机的交换机，这里采用主机名+交换机名（和RabbitMQ一样）
        exchangeName =  exchangeName + virtualHostName;
        try{
            synchronized (exchangeLocker){
                Exchange exitExchange = memoryDataCenter.getExchange(exchangeName);
                if(exitExchange!=null){
                    System.out.println("[VirtualHost] 交换机已经存在！exchangeName: "+exchangeName);
                    return true;//已经有了也算是创建好了
                }
                Exchange exchange = new Exchange();
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setType(exchangeType);
                exchange.setName(exchangeName);
                exchange.setOptions(argument);
                //创建好后，先写入硬盘，然后写入内存
                if(durable){
                    diskDataCenter.insertExchange(exchange);
                }
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成！exchangeName:  "+exchangeName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机创建失败！exchangeName: "+exchangeName);
            e.printStackTrace();
            return false;
        }
    }
    //删除交换机
    public boolean exchangeDelete(String exchangeName){
        exchangeName = exchangeName + virtualHostName;
        try{
            synchronized (exchangeLocker){
                Exchange deleteExchange = memoryDataCenter.getExchange(exchangeName);
                if(deleteExchange==null){
                    throw new MqException("[VirtualHost] 要删除的交换机不存在！exchangeName: "+exchangeName);
                }
                if(deleteExchange.isDurable()){
                    diskDataCenter.deleteExchange(deleteExchange);
                }
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 删除交换机成功！exchangeName: "+exchangeName);
            }
            return true;
        } catch (MqException e) {
            System.out.println("[VirtualHost] 删除交换机失败！exchangeName: "+exchangeName);
            e.printStackTrace();
            return false;
        }
    }
    //创建队列
    public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,
                                Map<String,Object> options){
        queueName = queueName + virtualHostName;
        try{
            synchronized (queueLocker){
                MQqueue exitQueue = memoryDataCenter.getMQqueue(queueName);
                if(exitQueue!=null){
                    System.out.println("[VirtualHost] 队列已经存在！queueName: "+queueName);
                    return true;
                }
                MQqueue queue = new MQqueue();
                queue.setExclusive(exclusive);
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setAutoDelete(autoDelete);
                queue.setOptions(options);
                if(durable){
                    diskDataCenter.insertQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName = " + queueName);
            }
            return true;
        } catch (IOException e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }
    //删除队列
    public boolean queueDelete(String queueName){
        queueName = queueName + virtualHostName;
        try{
            synchronized (queueLocker){
                MQqueue delQueue = memoryDataCenter.getMQqueue(queueName);
                if(delQueue==null){
                    throw new MqException("[VirtualHost] 队列不存在!无法删除！ queueName = " +queueName);
                }
                if(delQueue.isDurable()){
                    diskDataCenter.deleteQueue(delQueue);
                }
                memoryDataCenter.deleteMQqueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功! queueName = " + queueName);
            }
            return true;
        } catch (MqException | IOException e) {
            System.out.println("[VirtualHost] 删除队列失败! queueName = " + queueName);
            e.printStackTrace();
            return false;
        }
    }
    //创建绑定
    public boolean queueBind(String queueName,String exchangeName,String bindingKey){
        queueName = queueName + virtualHostName;
        exchangeName =  exchangeName + virtualHostName;
        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    Binding exitBinding = memoryDataCenter.getBinding(exchangeName,queueName);
                    if(exitBinding!=null){
                        throw new MqException("[VirtualHost] 绑定已经存在！");
                    }
                    if(!router.checkBindingKey(bindingKey)){
                        throw new MqException("[VirtualHost] bindingKey非法！"+bindingKey);
                    }
                    //创建binding
                    Binding binding = new Binding();
                    binding.setBindKey(bindingKey);
                    binding.setQueueName(queueName);
                    binding.setExchangeName(exchangeName);
                    //验证queue和exchange
                    MQqueue queue = memoryDataCenter.getMQqueue(queueName);
                    if(queue==null){
                        throw new MqException("[VirtualHost] 队列不存在! queueName = " + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange==null){
                        throw new MqException("[VirtualHost] 交换机不存在! exchangeName = " + exchangeName);
                    }
                    if(queue.isDurable()&&exchange.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功! exchangeName = " + exchangeName
                    + ", queueName=" + queueName);
            return true;
        } catch (MqException e) {
            System.out.println("[VirtualHost] 绑定创建失败! exchangeName = " + exchangeName
                    + ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    //删除队列
    public boolean queueUnbinding(String queueName,String exchangeName){
        queueName =  queueName + virtualHostName;
        exchangeName =  exchangeName + virtualHostName;
        try{
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    Binding binding = memoryDataCenter.getBinding(exchangeName,queueName);
                    if(binding==null){
                        throw new MqException("[VirtualHost] 删除绑定失败! 绑定不存在! exchangeName = " + exchangeName + ", queueName = " + queueName);
                    }
                    diskDataCenter.deleteBinding(binding);
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 删除绑定成功！");
                }
            }
            return true;
        } catch (MqException e) {
            System.out.println("[VirtualHost] 删除绑定失败!");
            e.printStackTrace();
            return false;
        }
    }
//发送消息到指定的交换机/队列
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties,byte[] body){
        try{
            exchangeName = exchangeName + virtualHostName;
            if(!router.checkRoutingKey(routingKey)){
                throw new MqException("[VirtualHost] routingKey非法！");
            }
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange==null){
                throw new MqException("[VirtualHost] exchange不存在！exchangeName: "+exchangeName);
            }
            if(exchange.getType()==ExchangeType.DIRECT){
                //直接发给指定的队列
                String queueName = routingKey + virtualHostName;
                MQqueue queue = memoryDataCenter.getMQqueue(queueName);
                if(queue==null){
                    throw new MqException("[VirtualHost] 对列不存在！queueName: "+queueName);
                }
                //构建消息
                Message message = Message.createMessageWithId(basicProperties,body,routingKey);
                sendMessage(queue,message);
            }else {
                //按照FANOUT和TOPIC来转发消息
                ConcurrentHashMap<String,Binding> bindings = memoryDataCenter.getBindings(exchangeName);
                //fanout直接给所有队列发消息
                for(Map.Entry<String,Binding> entry:bindings.entrySet()){
                    Binding binding = entry.getValue();
                    MQqueue queue = memoryDataCenter.getMQqueue(binding.getQueueName());
                    if(queue==null){
                        System.out.println("[VirtualHost] basicPublish 发送消息时, 发现队列不存在! queueName = " + binding.getQueueName());
                        continue;
                        //一个队列不存在不要影响后面的转发
                    }
                    Message message = Message.createMessageWithId(basicProperties,body,routingKey);
                    if(!router.route(exchange.getType(),binding,message)){
                        continue;
                    }
                    sendMessage(queue,message);
                }
            }
            return true;
        } catch (MqException | IOException | InterruptedException e) {
            System.out.println("[VirtualHost] 消息发送失败!");
            e.printStackTrace();
            return false;
        }
    }
//发送消息就是将消息写到硬盘和内存上
    private void sendMessage(MQqueue queue, Message message) throws IOException, MqException, InterruptedException {
        int deliverMode = message.getBasicProperties().getDeliverMode();
        if(deliverMode==2){
            diskDataCenter.sendMessage(queue,message);
        }
        memoryDataCenter.sendMessage(queue,message);
        consumerManager.notifyConsume(queue.getName());
    }
    //订阅信息
    //添加一个队列的订阅者，当队列收到消息后，就要把消息推给对应的订阅者
    //consumeTag： 消费者的身份标识
    //autoAck：消息被消费完成后，应答的方式，true为自动应答，false为手动应答
    //consumer：是一个回调函数，此处类型设定为函数式接口，调用时写作lambda样子
    public boolean basicConsume(String consumerTag,String queueName,boolean autoAck,Consumer consumer){
        queueName =   queueName + virtualHostName;
        try{
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsume成功！queueName="+queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] basicConsume 失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }
    public boolean basicAck(String queueName,String messageId){
        queueName = queueName + virtualHostName;
        try{
            Message message = memoryDataCenter.getMessage(messageId);
            if (message == null) {
                throw new MqException("[VirtualHost] 要确认的消息不存在! messageId=" + messageId);
            }
            MQqueue queue = memoryDataCenter.getMQqueue(queueName);
            if (queue == null) {
                throw new MqException("[VirtualHost] 要确认的队列不存在! queueName=" + queueName);
            }
            // 删除硬盘上的数据
            if (message.getBasicProperties().getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            //删除消息中心的数据
            memoryDataCenter.deleteMessage(message);
            memoryDataCenter.deleteMessageWaitAck(queueName,messageId);
            System.out.println("[VirtualHost] basicAck 成功! 消息被成功确认! queueName=" + queueName
                    + ", messageId=" + messageId);
            return true;
        } catch (MqException | ClassNotFoundException | IOException e) {
            System.out.println("[VirtualHost] basicAck 失败! 消息确认失败! queueName=" + queueName
                    + ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }



}
